package com.prac.mq.controller;

import cn.hutool.json.JSONUtil;
import com.prac.mq.model.MessageModel;

import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/mq")
@Slf4j
public class MqController {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @RequestMapping("test")
    public void send(HttpServletRequest request) {
        /**
         * 异步消息
         */
        rocketMQTemplate.syncSend("boot-topic", "boot message消息体");
        log.info("投递成功");
        rocketMQTemplate.asyncSend("boot-topic", "async message消息体", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("投递成功，消息状态：{}", sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("投递失败，{}", throwable.getMessage());
            }
        });
        /**
         * 延迟消息
         */
        Message<String> build = MessageBuilder.withPayload("延迟消息消息体").setHeader(RocketMQHeaders.KEYS, "aaa").build();
        rocketMQTemplate.syncSend("boot-topic", build, 1000 * 10, 3);

        /**
         * 顺序消息
         */
        List<MessageModel> messageList = Arrays.asList(
                new MessageModel("1001", "user1", "u1-顺序消息2"),
                new MessageModel("1001", "user1", "u1-顺序消息1"),

                new MessageModel("1002", "user2", "u2-顺序消息1"),
                new MessageModel("1002", "user2", "u2-顺序消息2"),
                new MessageModel("1002", "user2", "u2-顺序消息3"),

                new MessageModel("1003", "user3", "u3-顺序消息1"),
                new MessageModel("1003", "user3", "u3-顺序消息2"),
                new MessageModel("1003", "user3", "u3-顺序消息3")
        );
        messageList.forEach(t -> {
            rocketMQTemplate.syncSendOrderly("boot-orderly-topic", JSONUtil.toJsonStr(t), t.getOrderId());
        });

    }

    @GetMapping("tran")
    public void testTrans(@RequestParam String name) {
        rocketMQTemplate.sendMessageInTransaction(
                "boot-trans",
                MessageBuilder.withPayload(name).build(),
                (Consumer<String>) s -> {
                    //演示本地事务执行失败，回滚事务
                    // int i = 1 / 0;
                    log.info("事务消息执行，消息体：{}", s);
                });
    }

    @Component
    @RocketMQMessageListener(topic = "boot-trans", consumerGroup = "boot-consumer-group")
    public static class TransListener implements RocketMQListener<MessageExt> {
        @Override
        public void onMessage(MessageExt message) {
            log.info("消息消费：{},trans-msgId:{}", new String(message.getBody()), message.getTransactionId());
        }
    }

    @Component
    @RocketMQMessageListener(topic = "boot-topic", consumerGroup = "boot-consumer-group")
    public static class SimpleListener implements RocketMQListener<MessageExt> {
        /**
         * 不报错就签收；报错就拒收 重试
         */
        @Override
        public void onMessage(MessageExt message) {
            log.info("消息消费：{}", new String(message.getBody()));
        }
    }

    @Component
    @RocketMQMessageListener(
            topic = "boot-orderly-topic", consumerGroup = "boot-orderly-consumerGroup",
            consumeMode = ConsumeMode.ORDERLY, maxReconsumeTimes = 5
    )
    public static class OrderlyListener implements RocketMQListener<MessageExt> {
        @Override
        public void onMessage(MessageExt message) {
            log.info("消息消费：{}", new String(message.getBody()));
        }
    }

}
